package com.atguigu.dga.lineage.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.dga.ds.bean.TDsTaskDefinition;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.ds.service.TDsTaskDefinitionService;
import com.atguigu.dga.ds.service.TDsTaskInstanceService;
import com.atguigu.dga.lineage.bean.GovernanceLineageTable;
import com.atguigu.dga.lineage.mapper.GovernanceLineageTableMapper;
import com.atguigu.dga.lineage.service.GovernanceLineageTableService;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.atguigu.dga.util.SqlParser;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.Getter;
import lombok.Setter;
import org.antlr.runtime.tree.Tree;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.velocity.runtime.parser.node.ASTOrNode;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2023-11-27
 */
@Service
@DS("dga")
public class GovernanceLineageTableServiceImpl extends ServiceImpl<GovernanceLineageTableMapper, GovernanceLineageTable> implements GovernanceLineageTableService {

    //1   要提取哪些表 -->  元数据   tableMetaInfoList

    //2   当天任务实例 的   task的任务定义 sql

    // 3   解析sql --> 提取来源表
        //通过 遍历sql 最后得到  -->    来源表关系map    sourceMap  Map<String,List<String>>    key 输出表的表名  values 来源表的集合
                                 // 输出表关系map    sinkMap  Map<String,List<String>>    key 来源表的表名  values  输出表的集合
    // 4  遍历元数据表
        //   从 来源表关系map 和 输出表关系map 中提取数据  来源表关系map->sourceTables    输出表关系map -->sinkTables

    //5保存  lineage_table


    @Autowired
    TableMetaInfoService tableMetaInfoService;

    @Autowired
    TDsTaskInstanceService tDsTaskInstanceService;

    @Autowired
    TDsTaskDefinitionService tDsTaskDefinitionService;

    public  void  extractLineageTable(String governanceDate) throws Exception {

        //0 清除当天内容
        remove(new QueryWrapper<GovernanceLineageTable>().eq("governance_date",governanceDate));


        //1  要提取哪些表 -->  元数据   tableMetaInfoList
        List<TableMetaInfo> tableMetaInfoList=tableMetaInfoService.getTableMetaInfoWithExtraList(governanceDate);
        //2   当天任务实例 的   task的任务定义 sql
        //把当天运行最后一次实例对应的任务定义取出来  map<表名,任务定义>
        Map<String, TDsTaskDefinition>  taskDefinitionMap=tDsTaskDefinitionService.getTaskDefinitionMap(governanceDate);

        // 3   解析sql --> 提取来源表
        //通过 遍历sql 最后得到  -->    来源表关系map    sourceMap  Map<String,List<String>>    key 输出表的表名  values 来源表的集合
        // 输出表关系map    sinkMap  Map<String,List<String>>    key 来源表的表名  values  输出表的集合
        HashMap<String, Collection<String>> sourceTableMap = new HashMap<>();// 来源表关系map
        HashMap<String, Collection<String>> sinkTableMap = new HashMap<>();// 输出表关系map

        for (TDsTaskDefinition tDsTaskDefinition : taskDefinitionMap.values()) {
            String sql = tDsTaskDefinition.getSql();
            if(tDsTaskDefinition.getName().equals("gmall.ads_sku_cart_num_top3_by_cate")){
                System.out.println(111);
            }
            if (sql!=null){
                //解析sql 目的：获取来源表的集合
                // 调用解析sql工具 ， 制作一个节点处理器：目的就是提取来源表
                SourceTableDispatcher sourceTableDispatcher = new SourceTableDispatcher();
                String[] tableNameWithSchema = tDsTaskDefinition.getName().split("\\.");
                String sinkTableSchemaName = tableNameWithSchema[0];
                sourceTableDispatcher.setDefaultSchemaName(sinkTableSchemaName);

                SqlParser.parse(sql,sourceTableDispatcher);

                Set<String> sourceTableSet = sourceTableDispatcher.getSourceTableSet();
                Set<String> subqueryTableSet = sourceTableDispatcher.getSubqueryTableSet();
                //去除子查询
                sourceTableSet.removeAll(subqueryTableSet); //剩下真实表
                //去除自己
                sourceTableSet.remove(tDsTaskDefinition.getName());


                String sinkTableName=tDsTaskDefinition.getName();
                sourceTableMap.put(sinkTableName,sourceTableSet); //填来源表

                for (String sourceTableName : sourceTableSet) {
                    Collection<String> sinKTableNames = sinkTableMap.get(sourceTableName);
                    if(sinKTableNames==null){
                        sinKTableNames=new HashSet<String>();
                    }
                    sinKTableNames.add(sinkTableName);
                    sinkTableMap.put( sourceTableName,sinKTableNames);//填输出表
                }


            }


        }
        // 4  遍历元数据表
        //   从 来源表关系map 和 输出表关系map 中提取数据  来源表关系map->sourceTables    输出表关系map -->sinkTables

        List<GovernanceLineageTable> lineageTableList=new ArrayList<>();
        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
            if(tableMetaInfo.getTableName().equals("ads_sku_cart_num_top3_by_cate")){
                System.out.println(1111);
            }

            GovernanceLineageTable governanceLineageTable = new GovernanceLineageTable();
            governanceLineageTable.setTableName(tableMetaInfo.getTableName());
            governanceLineageTable.setSchemaName(tableMetaInfo.getSchemaName());

            String tableNameWithSchema=tableMetaInfo.getSchemaName()+"."+tableMetaInfo.getTableName();
            Collection<String> sinkTables = sinkTableMap.get(tableNameWithSchema); //从输出表map中提取
            if(sinkTables!=null&&sinkTables.size()>0){
                governanceLineageTable.setSinkTables(JSON.toJSONString(sinkTables ));
            }


            Collection<String> sourceTables = sourceTableMap.get(tableNameWithSchema); //从来源表map中提取
            if(sourceTables!=null&&sourceTables.size()>0) {
                governanceLineageTable.setSourceTables(JSON.toJSONString(sourceTables));
            }

            governanceLineageTable.setGovernanceDate(governanceDate);
            governanceLineageTable.setCreateTime(new Date());

            lineageTableList.add(governanceLineageTable);

        }


        //5保存  lineage_table
        saveBatch(lineageTableList);


    }


    class  SourceTableDispatcher implements Dispatcher{

        @Setter
        String defaultSchemaName=null;

        @Getter
        Set<String>  sourceTableSet=new HashSet<>();  //来源表 （含真实表和 子查询表)

        @Getter
        Set<String>  subqueryTableSet=new HashSet<>(); //子查询表名


        @Override
        public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs) throws SemanticException {
            ASTNode astNode = (ASTNode) nd;

            if(astNode.getType()== HiveParser.TOK_TABREF){
                ASTNode tableOrColNode =(ASTNode) astNode.getChild(0);
                if(tableOrColNode.getChildren().size()>1){//带库名
                      String schemaName = tableOrColNode.getChild(0).getText();
                      String tableName = tableOrColNode.getChild(1).getText();
                      sourceTableSet.add(schemaName+"."+tableName);
                }else{                                    // 不带库名
                    String tableName = tableOrColNode.getChild(0).getText();
                      sourceTableSet.add(defaultSchemaName+"."+tableName);
                }

            }else if(astNode.getType()== HiveParser.TOK_SUBQUERY&&astNode.getChildCount()==2){  //收集子查询名
                 ASTNode  subqueryNameNode = (ASTNode)astNode.getChild(1);
                 String subqueryName = subqueryNameNode.getText();
                 subqueryTableSet.add(defaultSchemaName+"."+subqueryName);
            }

            return null;
        }
    }

}
